package com.jthinking.service.kafka.delay.queue.listener;

import com.jthinking.service.kafka.delay.queue.storage.StorageService;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.Calendar;
import java.util.List;

@Component
public class TimeoutListener {

    private static final Logger LOGGER = LoggerFactory.getLogger(TimeoutListener.class);

    @Resource
    private StorageService storageService;
    @Resource
    private KafkaTemplate<String, String> stringKafkaTemplate;

    @KafkaListener(
            topics = {"delay-queue-timeout"},
            properties = {
                    "value.deserializer:org.apache.kafka.common.serialization.LongDeserializer"
            }
    )
    public void listen(List<Long> list) {
        for (Long currentTimeMillis : list) {
            Calendar instance = Calendar.getInstance();
            instance.setTimeInMillis(currentTimeMillis);
            int year = instance.get(Calendar.YEAR);
            String month = String.format("%02d", instance.get(Calendar.MONTH) + 1);
            String day = String.format("%02d", instance.get(Calendar.DAY_OF_MONTH));
            String hour = String.format("%02d", instance.get(Calendar.HOUR_OF_DAY));
            String minute = String.format("%02d", instance.get(Calendar.MINUTE));
            String minuteStr = year + month + day + hour + minute;
            int secondIndex = instance.get(Calendar.SECOND);

            LOGGER.info("Timeout! minuteStr: {}, secondIndex: {}", minuteStr, secondIndex);
            List<byte[]> result = storageService.get(minuteStr, secondIndex);
            for (byte[] r : result) {
                stringKafkaTemplate.send("delay-queue-result", new String(r));
            }
        }
    }

}
